package com.simple.redissonrpc.core;

import cn.hutool.extra.spring.SpringUtil;
import com.simple.redissonrpc.config.GameConfig;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.api.listener.MessageListener;
import org.redisson.codec.SerializationCodec;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class RpcService {

    public static final String RPC_PROVIDER = "RPC_PROVIDER_";

    public static final int TIME_OUT_SEC = 10;

    private final Map<Long, CallContext<AbsRpcAnswer>> seq2CallContextMap = new ConcurrentHashMap<>();

    private final Map<String, RpcCmdExecutor> className2CmdExecutorMap = new HashMap<>();

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private GameConfig gameConfig;

    public void init() {
        initTopic();

        initHandler();

        log.info("rpc init done. serverId={}", gameConfig.getServerId());
    }


    private void initTopic() {
        String topic = RPC_PROVIDER + gameConfig.getServerId();

        RTopic rTopic = redissonClient.getTopic(topic, new SerializationCodec());
        rTopic.addListener(ServerInternalMessage.class, new MessageListener<>() {
            @Override
            public void onMessage(CharSequence charSequence, ServerInternalMessage msg) {
                // 2个进程的serverId一样，这样子会造成同一个事件2个监听，过滤掉
                if (ignoreMsg(msg)) {
                    return;
                }

                // send和call消息
                if (msg.getType() == RpcMsgType.SEND) {
                    // 业务处理
                    callProviderMethod(msg, false);
                } else if (msg.getType() == RpcMsgType.CALL) {
                    // 业务处理
                    callProviderMethod(msg, true);
                } else if (msg.getType() == RpcMsgType.RESPONSE) {  // 返回消息
                    CallContext<AbsRpcAnswer> context = seq2CallContextMap.remove(msg.getCallbackId());
                    if (context != null) {
                        context.getFuture().complete((AbsRpcAnswer) msg.getData());
                    } else {
                        log.error("找不到callbackId={}", msg.getCallbackId());
                    }
                }
            }
        });
    }

    private void initHandler() {
        Collection<Object> values = SpringUtil.getApplicationContext().getBeansWithAnnotation(Controller.class).values();
        for (Object value : values) {
            Method[] declaredMethods = value.getClass().getDeclaredMethods();
            for (Method method : declaredMethods) {
                RpcMsgReceiver rpcMsgReceiverAnnotation = method.getAnnotation(RpcMsgReceiver.class);
                if (rpcMsgReceiverAnnotation != null) {
                    String simpleName = rpcMsgReceiverAnnotation.value().getSimpleName();
                    className2CmdExecutorMap.put(simpleName, new RpcCmdExecutor(method, value));
                }
            }
        }
    }

    private void callProviderMethod(ServerInternalMessage<IRpcInternalMsg> msg, boolean callFlag) {
        String className = msg.getClassName();

        try {
            RpcCmdExecutor rpcCmdExecutor = className2CmdExecutorMap.get(className);
            if (rpcCmdExecutor != null) {
                GameRpcMsgContext gameRpcMsgContext = new GameRpcMsgContext();
                gameRpcMsgContext.setFromServerId(msg.getFromServerId());
                gameRpcMsgContext.setCallbackId(msg.getSeqId());
                gameRpcMsgContext.setClassName(msg.getClassName());
                gameRpcMsgContext.setMsg(msg.getData());

                rpcCmdExecutor.invoke(gameRpcMsgContext);
            }
        } catch (Exception e) {
            if (callFlag) {
                response(msg.getFromServerId(), new ExceptionAnswer(), msg.getSeqId());
            }
            log.error("服务提供者出错msg={},error=", msg, e);
        }
    }

    private boolean ignoreMsg(ServerInternalMessage<IRpcInternalMsg> msg) {
        // serverId一样时。只允许自己处理自己
        if (msg.getFromServerId() == gameConfig.getServerId() && msg.getMachineId() != gameConfig.getMachineId()) {
            return true;
        }

        return false;
    }

    private <T extends AbsRpcAnswer> T getResponseMessage(long seqId, T ret, Class<T> clazz) throws Exception {
        if (ret.getClass().getSimpleName().equals(TimeOutAnswer.class.getSimpleName())) {
            seq2CallContextMap.remove(seqId);
            T t = clazz.getConstructor().newInstance();
            t.setErrorCode(RpcErrorCode.TIMEOUT);
            return t;
        } else if (ret.getClass().getSimpleName().equals(ExceptionAnswer.class.getSimpleName())) {
            seq2CallContextMap.remove(seqId);
            T t = clazz.getConstructor().newInstance();
            t.setErrorCode(RpcErrorCode.EXCEPTION);
            return t;
        } else {
            ret.setErrorCode(RpcErrorCode.OK);
        }
        return ret;
    }

    /**
     * 发送send消息
     *
     * @param toServerId
     * @param data
     */
    public <T extends IRpcInternalMsg> void send(int toServerId, T data) {
        RTopic rTopic = redissonClient.getTopic(RPC_PROVIDER + toServerId, new SerializationCodec());

        ServerInternalMessage<IRpcInternalMsg> msg = ServerInternalMessage
                .valueOf(gameConfig.getServerId(), data.getClass().getSimpleName(), data, RpcMsgType.SEND);

        rTopic.publish(msg);
    }

    /**
     * 发起syncCall
     *
     * @param toServerId 调用的哪个服务的
     * @param data
     * @param clazz
     * @param <T>
     * @return
     * @throws Exception
     */
    public <T extends AbsRpcAnswer, D extends IRpcInternalMsg> T syncCall(int toServerId, D data, Class<T> clazz) throws Exception {
        CompletableFuture<T> completableFuture = new CompletableFuture<>();

        RTopic rTopic = redissonClient.getTopic(RPC_PROVIDER + toServerId, new SerializationCodec());

        ServerInternalMessage<IRpcInternalMsg> msg = ServerInternalMessage
                .valueOf(gameConfig.getServerId(), data.getClass().getSimpleName(), data, RpcMsgType.CALL);

        CallContext<AbsRpcAnswer> context = new CallContext<>();
        context.setFuture((CompletableFuture<AbsRpcAnswer>) completableFuture);

        // 保存下上下文
        seq2CallContextMap.put(msg.getSeqId(), context);

        rTopic.publish(msg);

        // 同步等待
        T ret = completableFuture.completeOnTimeout((T) new TimeOutAnswer(), TIME_OUT_SEC, TimeUnit.SECONDS).get();
        return getResponseMessage(msg.getSeqId(), ret, clazz);
    }



    /**
     * 发起asyncCall
     */
    public <T extends AbsRpcAnswer, D extends IRpcInternalMsg> void asyncCall(int toServerId, D data, Class<T> clazz, Consumer<T> consumer) {
        CompletableFuture<T> completableFuture = new CompletableFuture<>();

        RTopic rTopic = redissonClient.getTopic(RPC_PROVIDER + toServerId, new SerializationCodec());

        ServerInternalMessage<IRpcInternalMsg> msg = ServerInternalMessage
                .valueOf(gameConfig.getServerId(), data.getClass().getSimpleName(), data, RpcMsgType.CALL);

        // 保存下上下文
        CallContext<AbsRpcAnswer> context = new CallContext<>();

        // 异步等待
        completableFuture //
                .completeOnTimeout((T) new TimeOutAnswer(), TIME_OUT_SEC, TimeUnit.SECONDS) // 超时检测
                .whenComplete((ret, e) -> { // 拿到结果
                    T responseMessage = null;
                    try {
                        responseMessage = getResponseMessage(msg.getSeqId(), ret, clazz);
                    } catch (Exception ex) {
                        log.error("ex=", ex);
                    }
                    consumer.accept(responseMessage);
                });
        context.setFuture((CompletableFuture<AbsRpcAnswer>) completableFuture);

        seq2CallContextMap.put(msg.getSeqId(), context);

        rTopic.publish(msg);
    }

    /**
     * 给call消息发送返回
     *
     * @param toServerId
     * @param data
     * @param
     */
    public <T extends IRpcInternalMsg> void response(int toServerId, T data, long callbackId) {
        RTopic rTopic = redissonClient.getTopic(RPC_PROVIDER + toServerId, new SerializationCodec());

        ServerInternalMessage<IRpcInternalMsg> msg = ServerInternalMessage
                .valueOf(gameConfig.getServerId(), data.getClass().getSimpleName(), data, RpcMsgType.RESPONSE);
        msg.setCallbackId(callbackId);

        rTopic.publish(msg);
    }
}
